"""
Case Type   : GMS_LOB
Case Name   : GMS_LOB write、append、read、getlength、isopen。gs_dump导出导入测试
Create At   : 2025/5/26
Owner       : cting66
Description :
Expect      :
History     :
"""

import os
import unittest

from yat.test import Node
from yat.test import macro

from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger


class CompatibilityTest01(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.common = Common()
        self.sh_primary = CommonSH('PrimaryDbAmode')
        self.user_node = Node('PrimaryDbAmode')
        self.constant = Constant()

    def test_gms_lob(self):
        self.log.info(f'----{os.path.basename(__file__)} start----')
        text = '--1. write、append函数导出导入测试;expect: 测abcABC123abc--'
        self.log.info(text)

        sql_cmd = f'''drop database if exists dbtest_dump;
                        drop database if exists dbtest_restore;
                        create database dbtest_dump ENCODING 'UTF-8';
                        create database dbtest_restore ENCODING 'UTF-8';
                   '''
        self.log.info(sql_cmd)
        sql_res = self.sh_primary.execut_db_sql(sql_cmd)
        self.log.info(sql_res)
        self.assertIn(self.constant.CREATE_DATABASE_SUCCESS, sql_res, '执行失败')

        sql_cmd = f"create or replace function cast_to_raw(strdata varchar2) returns raw " \
                  f"as 'select encode(cast(\$1 as bytea), ''hex'')::raw;'LANGUAGE SQL; "
        self.log.info(sql_cmd)
        sql_res = self.sh_primary.execut_db_sql(sql_cmd, dbname='dbtest_dump')
        self.log.info(sql_res)
        self.assertIn(self.constant.CREATE_FUNCTION_SUCCESS_MSG, sql_res, '执行失败')

        sql_cmd = f"create extension gms_lob;"
        self.log.info(sql_cmd)
        sql_res = self.sh_primary.execut_db_sql(sql_cmd, dbname='dbtest_dump')
        self.log.info(sql_res)
        self.assertIn(self.constant.create_extension_success, sql_res, '执行失败')

        sql_cmd = f'''create or replace function func_test() return text
is testvalue text;
declare
b1 blob :=cast_to_raw('测试ABC123');
amount INTEGER :=3;
off_set INTEGER :=4;
b2 blob :=cast_to_raw('abc');
begin
gms_lob.write(b1,amount,off_set,b2);
gms_lob.append(b1,b2);
testvalue :=to_char(b1);
return testvalue;
end;'''
        self.log.info(sql_cmd)
        sql_res = self.sh_primary.execut_db_sql(sql_cmd, dbname='dbtest_dump')
        self.log.info(sql_res)
        self.assertIn(self.constant.CREATE_FUNCTION_SUCCESS_MSG, sql_res, '执行失败')

        self.log.info('-----使用gs_dump备份数据-----')
        dump_res = self.sh_primary.exec_gs_dump('$GAUSSHOME/backup.sql',
                                                '-F p --include-extension',
                                                'dbtest_dump', get_detail=True)
        self.assertIn(f"dump database dbtest_dump successfully",
                      dump_res, '备份失败')

        self.log.info('-----使用gsql导入数据-----')
        cmd = f"source {macro.DB_ENV_PATH};" \
              f"gsql -d dbtest_restore -p {self.user_node.db_port} " \
              f"-U {self.user_node.ssh_user} " \
              f"-W {self.user_node.ssh_password} " \
              f"-f $GAUSSHOME/backup.sql"
        self.log.info(cmd)
        res = self.common.get_sh_result(self.user_node, cmd)
        self.log.info(res)
        self.assertIn('CREATE FUNCTION', res, 'gsql恢复失败')

        sql_cmd = f"select func_test();"
        self.log.info(sql_cmd)
        sql_res = self.sh_primary.execut_db_sql(sql_cmd, dbname='dbtest_restore')
        self.log.info(sql_res)
        self.assertIn('测abcABC123abc', sql_res, 'write、append函数导出导入测试失败')


        self.log.info(f'----{os.path.basename(__file__)} start----')
        text = '--2. read、getlength、isopen函数导入导出测试--'
        self.log.info(text)
        sql_cmd = f'''drop database if exists dbtest_dump;
                                drop database if exists dbtest_restore;
                                create database dbtest_dump ENCODING 'UTF-8';
                                create database dbtest_restore ENCODING 'UTF-8';
                           '''
        self.log.info(sql_cmd)
        sql_res = self.sh_primary.execut_db_sql(sql_cmd)
        self.log.info(sql_res)
        self.assertIn(self.constant.CREATE_DATABASE_SUCCESS, sql_res, '执行失败')

        sql_cmd = f"create or replace function cast_to_raw(strdata varchar2) returns raw " \
                  f"as 'select encode(cast(\$1 as bytea), ''hex'')::raw;'LANGUAGE SQL; "
        self.log.info(sql_cmd)
        sql_res = self.sh_primary.execut_db_sql(sql_cmd, dbname='dbtest_dump')
        self.log.info(sql_res)
        self.assertIn(self.constant.CREATE_FUNCTION_SUCCESS_MSG, sql_res, '执行失败')

        sql_cmd = f"create extension gms_lob;"
        self.log.info(sql_cmd)
        sql_res = self.sh_primary.execut_db_sql(sql_cmd, dbname='dbtest_dump')
        self.log.info(sql_res)
        self.assertIn(self.constant.create_extension_success, sql_res, '执行失败')

        sql_cmd = f'''create or replace procedure proc_test_1()
as
declare
lob1 blob:= cast_to_raw('1234567890');
lob2 clob:= '中文数据内容测试';
amount int := 2;
off_set int := 5;
buf varchar(20);
begin
gms_lob.open(lob1,gms_lob.lob_readwrite);
raise info 'INFO: lob1 isopen: %', gms_lob.isopen(lob1);
gms_lob.read(lob1,amount,off_set,buf);
raise info 'INFO: buf: %', to_char(buf);
raise info 'INFO: buf length: %', gms_lob.getlength(buf);
gms_lob.open(lob2,gms_lob.lob_readwrite);
raise info 'INFO: lob2 isopen: %', gms_lob.isopen(lob2);
gms_lob.read(lob2,amount,off_set,buf);
raise info 'INFO: buf: %', to_char(buf);
raise info 'INFO: buf length: %', gms_lob.getlength(buf);
end;'''
        self.log.info(sql_cmd)
        sql_res = self.sh_primary.execut_db_sql(sql_cmd, dbname='dbtest_dump')
        self.log.info(sql_res)
        self.assertIn(self.constant.CREATE_PROCEDURE_SUCCESS_MSG, sql_res, '执行失败')

        sql_cmd = f'''create or replace function func_test_1() return void
as
declare
lob1 blob:= cast_to_raw('1234567890');
lob2 clob:= '中文数据内容测试';
amount int := 2;
off_set int := 5;
buf varchar(20);
BEGIN
gms_lob.open(lob1,gms_lob.lob_readwrite);
raise info 'INFO: lob1 isopen: %', gms_lob.isopen(lob1);
gms_lob.read(lob1,amount,off_set,buf);
raise info 'INFO: buf: %', to_char(buf);
raise info 'INFO: buf length: %', gms_lob.getlength(buf);
gms_lob.open(lob2,gms_lob.lob_readwrite);
raise info 'INFO: lob2 isopen: %', gms_lob.isopen(lob2);
gms_lob.read(lob2,amount,off_set,buf);
raise info 'INFO: buf: %', to_char(buf);
raise info 'INFO: buf length: %', gms_lob.getlength(buf);
return;
end;'''
        self.log.info(sql_cmd)
        sql_res = self.sh_primary.execut_db_sql(sql_cmd, dbname='dbtest_dump')
        self.log.info(sql_res)
        self.assertIn(self.constant.CREATE_FUNCTION_SUCCESS_MSG, sql_res, '执行失败')

        self.log.info('-----使用gs_dump备份数据-----')
        dump_res = self.sh_primary.exec_gs_dump('$GAUSSHOME/backup.dmp',
                                                '-F c --include-extension',
                                                'dbtest_dump', get_detail=True)
        self.assertIn(f"dump database dbtest_dump successfully",
                      dump_res, '备份失败')

        self.log.info('-----使用gs_restore恢复数据-----')
        cmd = f'''source {macro.DB_ENV_PATH}
                gs_restore $GAUSSHOME/backup.dmp -d dbtest_restore '''
        self.log.info(cmd)
        res = self.common.get_sh_result(self.user_node, cmd)
        self.log.info(res)
        self.assertIn(self.constant.RESTORE_SUCCESS_MSG, res)

        sql_cmd = f"call proc_test_1();"
        self.log.info(sql_cmd)
        sql_res = self.sh_primary.execut_db_sql(sql_cmd, dbname='dbtest_restore')
        self.log.info(sql_res)
        self.assertIn('isopen: 1', sql_res, 'read、getlength、isopen函数导入导出测试失败')
        self.assertIn('buf: 3536', sql_res, 'read、getlength、isopen函数导入导出测试失败')
        self.assertIn('length: 4', sql_res, 'read、getlength、isopen函数导入导出测试失败')
        self.assertIn('isopen: 1', sql_res, 'read、getlength、isopen函数导入导出测试失败')
        self.assertIn('buf: 内容', sql_res, 'read、getlength、isopen函数导入导出测试失败')
        self.assertIn('length: 2', sql_res, 'read、getlength、isopen函数导入导出测试失败')

        sql_cmd = f"call func_test_1();"
        self.log.info(sql_cmd)
        sql_res = self.sh_primary.execut_db_sql(sql_cmd, dbname='dbtest_restore')
        self.log.info(sql_res)
        self.assertIn('isopen: 1', sql_res, 'read、getlength、isopen函数导入导出测试失败')
        self.assertIn('buf: 3536', sql_res, 'read、getlength、isopen函数导入导出测试失败')
        self.assertIn('length: 4', sql_res, 'read、getlength、isopen函数导入导出测试失败')
        self.assertIn('isopen: 1', sql_res, 'read、getlength、isopen函数导入导出测试失败')
        self.assertIn('buf: 内容', sql_res, 'read、getlength、isopen函数导入导出测试失败')
        self.assertIn('length: 2', sql_res, 'read、getlength、isopen函数导入导出测试失败')

    def tearDown(self):
        self.log.info('-----step4:清理环境; expect:成功-----')
        check_cmd = f'rm -rf $GAUSSHOME/backup.sql;' \
                    f'if [ -d $GAUSSHOME/backup.sql ]; ' \
                    f'then echo "exists"; else echo "not exists"; fi'
        self.log.info(check_cmd)
        res = self.common.get_sh_result(self.user_node, check_cmd)
        self.assertEqual('not exists', res, '文件删除失败')
        check_cmd = f'rm -rf $GAUSSHOME/backup.dmp;' \
                    f'if [ -d $GAUSSHOME/backup.dmp ]; ' \
                    f'then echo "exists"; else echo "not exists"; fi'
        self.log.info(check_cmd)
        res = self.common.get_sh_result(self.user_node, check_cmd)
        self.assertEqual('not exists', res, '文件删除失败')
        sql_cmd = f"drop database if exists dbtest_dump;"
        self.log.info(sql_cmd)
        sql_res = self.sh_primary.execut_db_sql(sql_cmd)
        self.assertIn('DROP DATABASE', sql_res, '数据库删除失败')
        sql_cmd = f"drop database if exists dbtest_restore;"
        self.log.info(sql_cmd)
        sql_res = self.sh_primary.execut_db_sql(sql_cmd)
        self.log.info(sql_res)
        self.assertIn('DROP DATABASE', sql_res, '数据库删除失败')
        self.log.info(f'-----{os.path.basename(__file__)} end-----')

